package com.weshare.apitest.source;

import com.weshare.apitest.beans.SensorReading;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Random;

/**
 * 自定义的数据源
 */
public class MySourceFunction implements SourceFunction<SensorReading> {

    private boolean running = true;

    @Override
    public void run(SourceContext<SensorReading> ctx) throws Exception {

        // 定义一个随机数发生器
        Random random = new Random();

        HashMap<String, Double> sensorTemMap = new HashMap<>();
        for (int i = 0; i < 10; i++) {
            sensorTemMap.put("sensor_" + (i + 1), 60 + random.nextGaussian() * 20);
        }


        while (running) {

            for (String sensorId : sensorTemMap.keySet()) {

                Double newTemp = sensorTemMap.get(sensorId) + random.nextGaussian();
                sensorTemMap.put(sensorId, newTemp);
                ctx.collect(new SensorReading(sensorId, System.currentTimeMillis(), newTemp));
            }

            Thread.sleep(1000L);
        }
    }

    @Override
    public void cancel() {

        running = false;
    }
}
